#ifndef __EXECUTOR_H__
#define __EXECUTOR_H__

#include <boost/coroutine2/all.hpp>
#include <boost/format.hpp>
#include <optional>
#include <memory>
#include <functional>
#include <map>
#include <list>
#include <utility>
#include <variant>
#include <optional>
#include <mutex>
#include <condition_variable>
#include <iostream>

#include "utils.hpp"

namespace coro {

    class Executor;
    class ContextInternal;
    template <typename ValType>
    class Context;
    template<typename ValType>
    using coro_t = boost::coroutines2::coroutine<std::optional<ValType>>;
    template <typename ValType>
    class Waker;

    class Executor {
    public:
        // Start an new corotine
        // spawn should only be run on the main thread
        // Fn should be void(Context<ValType> &)
        using EvtCb = std::function<void(int)>;
        using WakedCtxList = std::list<std::pair<ContextInternal *, std::optional<void *>>>;

        Executor();
        template <typename ValType, typename Fn>
        void spawn(Fn fn);
        void register_context(ContextInternal *ctx);
        template<typename ValType>
        void wake(Context<ValType> *ctx, ValType val);
        void run();
        void remove_ctx(std::pair<ContextInternal *, std::optional<void *>> &pair);
        template<typename Fn>
        int io_schedule(int fd, int events, Fn fn);

    private:
        std::list<ContextInternal *>            m_ctxs;
        WakedCtxList                            m_waked_ctxs;
        std::mutex                              m_lock;
        int                                     m_epoll_fd;
        int                                     m_event_fd;
        std::map<int, std::list<std::pair<int, EvtCb>>> m_polling_fds;

        int io_schedule_internal(int fd, int events, EvtCb &fn);
        void handle_events(int fd, int event);
    };

    // Waker should be moved only
    template <typename ValType>
    class WakerImpl {
    protected:
        void wake_impl(ValType val);
    private:
        Context<ValType>    *m_ctx;
        Executor            *m_executor;

        WakerImpl(Context<ValType> *ctx, Executor *executor)
            : m_ctx(ctx), m_executor(executor)
        {}
        friend class Waker<ValType>;
    };

    // Waker::wake should be called only once
    template <typename ValType>
    class Waker {
    public:
        Waker() = default;
        Waker(Waker &&other)
        {
            m_impl.release();
            m_impl.swap(other.m_impl);
        }
        Waker(Waker &other)
        {
            m_impl.release();
            m_impl.swap(other.m_impl);
        }
        Waker& operator=(Waker &&other)
        {
            m_impl.release();
            m_impl.swap(other.m_impl);
            return *this;
        }
        Waker(Context<ValType> *ctx, Executor *executor)
            : m_impl(new WakerImpl<ValType>(ctx, executor))
        {}
        void wake(ValType val)
        {
            m_impl->wake_impl(val);
            // Now we should release this waker
            m_impl.reset();
        }
    private:
        std::unique_ptr<WakerImpl<ValType>> m_impl;
    };

    class ContextInternal {
    public:
        virtual void wake(std::optional<void *> val) = 0;
        virtual ~ContextInternal() = default;
        virtual explicit operator bool() = 0;
        virtual bool operator!() = 0;
        virtual void free_val(void *val) = 0;
    };

    template <typename ValType>
    class Context : public ContextInternal {
    public:
        Context(typename coro_t<ValType>::push_type *push_type,
                typename coro_t<ValType>::pull_type &pull_type,
                Executor *executor)
            : m_push_type(push_type), m_pull_type(pull_type),
              m_executor(executor)
        {}

        ~Context()
        {
            delete m_push_type;
        }

        Waker<ValType> waker()
        {
            return Waker<ValType>(this, m_executor);
        }

        // Switch context and wait for result
        std::optional<ValType> yield()
        {
            // Switch to parent context...
            m_pull_type();
            // ...and then fetch the data
            return m_pull_type.get();
        }

        void wake(std::optional<void *> val) override
        {
            if (val) {
                ValType *_val = (ValType *)val.value();
                (*m_push_type)(std::optional<ValType>(*_val));
            } else {
                (*m_push_type)(std::optional<ValType>());
            }
        }

        explicit operator bool() override
        {
            return bool(*m_push_type);
        }

        bool operator!() override
        {
            return !bool(*this);
        }

        void free_val(void *val) override
        {
            ValType *_val = (ValType *)val;
            delete _val;
        }

        Executor *get_executor()
        {
            return m_executor;
        }

    private:
        typename coro_t<ValType>::push_type *m_push_type;
        typename coro_t<ValType>::pull_type &m_pull_type;
        Executor                            *m_executor;
    };
};

#include "executor.tpp"
#endif
